package com.cl.ks.flow.handler;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.cl.ks.entity.KsFilterStatistics;
import com.cl.ks.entity.KsFlowGraphLog;
import com.cl.ks.service.*;
import com.cl.ks.utils.ExceptionUtil;
import com.cl.spark.dto.SparkResult;
import com.cl.ks.entity.ClueAutoResult;
import com.cl.ks.entity.KsRule;
import com.cl.ks.flow.KsFlowGraphHandler;
import com.cl.ks.flow.base.BaseFlowNodeHandler;
import com.cl.ks.flow.enums.NodeHandlerEnum;
import com.cl.ks.flow.util.RuleUtil;
import com.cl.ks.flow.util.StaticCollection;
import com.querydsl.jpa.impl.JPAQueryFactory;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.stream.Collectors;

import static com.cl.ks.constant.QTableConstant.*;


@Component
public class KsRuleInvokeHandler extends BaseFlowNodeHandler {

    @Autowired
    private RuleUtil ruleUtil;
    @Autowired
    private ClueAutoResultService clueAutoResultService;
    @Autowired
    private JPAQueryFactory queryFactory;
    @Autowired
    private KsRuleService ksRuleService;
    @Autowired
    private KsFlowGraphLogService ksFlowGraphLogService;
    @Autowired
    SparkSession sparkSession;
    @Autowired
    private KsFilterStatisticsService ksFilterStatisticsService;


    @Override
    public SparkResult process(BaseFlowNodeHandler.ProcessParam processParam) {
        SparkResult originalSparkResult = processParam.getSparkNode().getInputSparkResultList().values().stream().findFirst().get();
        if (originalSparkResult.getCount() == 0) {
            return originalSparkResult;
        }

        List<Dataset<Row>> datasetList = new ArrayList<>();

        Integer ksFlowGraphId = processParam.getKsFlowGraph().getId();
        String filterName = processParam.getKsFlowGraph().getName();
        Map<String, Object> expression = processParam.getSparkNode().getNodeExpression();

        JSONArray ruleIdStringList = JSONArray.parseArray(JSONArray.toJSONString(expression.get("ruleIdList")));
        List<String> ruleCodeList = new ArrayList<>();
        List<Integer> ruleIdList = new ArrayList<>();
        for (int i = 0; i < ruleIdStringList.size(); i++) {
            String ruleIdString = ruleIdStringList.getString(i);
            ruleCodeList.add(ruleIdString);
            if (ruleIdString.contains(".")) {
                String[] ruleIdRange = ruleIdString.split("\\.");
                int ruleId = Integer.parseInt(ruleIdRange[0]);
                if (!ruleIdList.contains(ruleId)) {
                    ruleIdList.add(ruleId);
                }
            } else {
                Integer ruleId = Integer.valueOf(ruleIdString);
                if (!ruleIdList.contains(ruleId)) {
                    ruleIdList.add(ruleId);
                }
            }
        }

        List<KsRule> ksRuleList = new ArrayList<>();
        List<KsRule> ksRuleListByCode = queryFactory.selectFrom(qKsRule).where(qKsRule.ruleCode.in(ruleCodeList)).fetch();
        List<KsRule> ksRuleListById = queryFactory.selectFrom(qKsRule).where(qKsRule.ruleCode.isNull().and(qKsRule.id.in(ruleIdList))).fetch();
        ksRuleList.addAll(ksRuleListByCode);
        ksRuleList.addAll(ksRuleListById);
        if (ksRuleList.isEmpty()) {
            throw new RuntimeException(ruleIdStringList + " 规则不存在");
        }
        processParam.getKsFlowGraph().getRuleIdList().addAll(ksRuleList.stream().map(KsRule::getId).map(String::valueOf).collect(Collectors.toList()));

        List<String> dataTypeCodeList = queryFactory.selectDistinct(qRlRuleField.dataTypeCode).from(qRlRuleField).where(qRlRuleField.ruleId.in(ruleIdList)).fetch();
        processParam.getKsFlowGraph().getKsDataTypeCodeList().addAll(dataTypeCodeList);
        long c1 = System.currentTimeMillis();
        KsFlowGraphLog ksFlowGraphLog = processParam.getKsFlowGraphLog();
        int index = 0;
        int ruleSize = ksRuleList.size();
        for (KsRule rule : ksRuleList) {
            try {
                ksFlowGraphLog.appendMsg(++index + "/" + ruleSize + "-规则 " + rule.getId() + " " + rule.getRuleName() + ":");
                Optional<KsRule> ksRuleOptional = ksRuleService.findById(rule.getId());
                if (!ksRuleOptional.isPresent()) {
                    break;
                }

                KsRule ksRule = ksRuleOptional.get();

                SparkResult sparkResult = ruleUtil.handleRule(ksRule, originalSparkResult);
                Dataset<Row> dataset = sparkResult.getDataset();

                if (!dataset.isEmpty()) {
                    datasetList.add(dataset);

                    processParam.getKsFlowGraph().setRuleDataTotal((int) (processParam.getKsFlowGraph().getRuleDataTotal() + sparkResult.getCount()));
                    String pushClueAutoResultFlag = String.valueOf(expression.get("pushClueAutoResultFlag"));
                    if ("是".equals(pushClueAutoResultFlag) || "1".equals(pushClueAutoResultFlag)) {
                        StaticCollection.newClueAutoResultList(ksFlowGraphId);

                        String power = processParam.getKsFlowGraph().getResultPower() + "," + processParam.getKsFlowGraph().getPower();

                        Integer finalRuleId = ksRule.getId();
                        String finalRuleName = ksRule.getRuleName();
                        dataset.foreach(item -> {
                            ClueAutoResult clueAutoResult = new ClueAutoResult();
                            clueAutoResult.setRuleId(finalRuleId);
                            clueAutoResult.setRuleName(finalRuleName);
                            clueAutoResult.setFilterId(ksFlowGraphId);
                            clueAutoResult.setFilterName(filterName);
                            clueAutoResult.setDbCode(item.getAs("DB_CODE"));
                            clueAutoResult.setTableName(item.getAs("TABLE_NAME"));
                            String id = "";
                            if (JSONObject.parseObject(item.json()).containsKey("id")) {
                                id = item.getAs("id").toString();
                            }
                            clueAutoResult.setSourceId(id);
                            String msg = "";
                            if (!Objects.isNull(item.getAs("msg"))) {
                                msg = item.getAs("msg").toString();
                            }
                            clueAutoResult.setReason(msg);
                            clueAutoResult.setTips(msg);
                            clueAutoResult.setPower(RuleUtil.powerConvert(power, item));
                            clueAutoResult.setResultPower(RuleUtil.powerConvert(power, item));
                            StaticCollection.getClueAutoResultList(ksFlowGraphId).add(clueAutoResult);
                        });
                    }
                }
                try {
                    KsFilterStatistics ksFilterStatistics = queryFactory.selectFrom(qKsFilterStatistics).where(qKsFilterStatistics.ruleId.eq(ksRule.getId())).orderBy(qKsFilterStatistics.id.desc()).fetchFirst();
                    ksFilterStatistics.setClueResultTotal(ksFilterStatistics.getClueResultTotal() + StaticCollection.getClueAutoResultList(ksFlowGraphId).size());
                    ksFilterStatisticsService.save(ksFilterStatistics);
                }catch (Exception e) {
                   e.printStackTrace();
                }
                ksFlowGraphLog.appendMsg("线索量/数据量：" + dataset.count() + "/" + originalSparkResult.getCount());
            } catch (Exception e) {
                e.printStackTrace();
                processParam.getKsFlowGraphLog().appendMsg(ExceptionUtil.getExceptionChainAsString(e));
            } finally {
                ksFlowGraphLog.appendMsg(";");
                ksFlowGraphLogService.save(ksFlowGraphLog);
            }
        }

        ClueAutoResult deleteParam = new ClueAutoResult();
        deleteParam.setFilterId(ksFlowGraphId);
        deleteParam.setFilterName(filterName);
        clueAutoResultService.deleteAll(deleteParam);
        clueAutoResultService.saveAllPartition(StaticCollection.getClueAutoResultList(ksFlowGraphId), 1000, processParam.getKsFlowGraph().getAuth());
        StaticCollection.removeClueAutoResultList(ksFlowGraphId);
        System.out.println("#" + (System.currentTimeMillis() - c1));

        if (datasetList.isEmpty()) {
            return SparkResult.success(sparkSession.createDataset(new ArrayList<>(), originalSparkResult.getDataset().encoder()));
        }
        return SparkResult.success(datasetList.stream().reduce(Dataset::unionAll).orElse(null));
    }

    @Override
    public NodeHandlerEnum getType() {
        return NodeHandlerEnum.KS_RULE_INVOKE;
    }
}
